package com.ds.infra.im.server;

import com.ds.infra.im.codec.CustomProtocolDecoder;
import com.ds.infra.im.codec.CustomProtocolEncoder;
import com.ds.infra.im.constant.Command;
import com.ds.infra.im.model.*;
import com.ds.infra.im.protocol.ProtocolMsg;
import com.ds.infra.im.util.JsonUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 作用：Netty服务端，主要负责客户端的连接、心跳检测、消息转发等功能。
 *
 * @author WeiShaoying
 */
public class NettyServer {

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
    private static final int PORT = 8888;

    // 客户端连接管理器
    private static final Map<String, ChannelHandlerContext> clientMap = new ConcurrentHashMap<>();

    public static void main(String[] args) {
        // 创建两个EventLoopGroup，bossGroup处理连接请求
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // workerGroup 处理业务请求
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            // 创建服务端启动引导类
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)// 设置TCP连接队列长度
                    .childOption(ChannelOption.SO_KEEPALIVE, true)// 设置TCP长连接，一般如果两个小时内没有数据的通信时，TCP会自动发送一个活动探测数据报文
                    .childOption(ChannelOption.TCP_NODELAY, true)// 关闭延迟发送，即TCP_NODELAY选项，禁用Nagle算法
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            // 添加空闲状态检测处理器，心跳检测机制，30秒内如果没有数据发送，则发送心跳消息，发现对方无回应时，主动断开连接，避免浪费服务端资源，维护无效的连接
                            pipeline.addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS));

                            // 添加自定义协议编解码器
                            /*
                             * 数据进入流程：ByteBuf -> Decoder -> 业务处理器
                             * 数据发出流程：业务处理器 -> Encoder -> ByteBuf
                             * Inbound | Socket -> Decoder -> Handler1 -> Handler2 -> ... |
                             * Outbound| ... <- Handler2 <- Handler1 <- Encoder <- Socket |
                             */
                            pipeline.addLast(new CustomProtocolDecoder());
                            pipeline.addLast(new CustomProtocolEncoder());

                            // 添加业务处理器
                            pipeline.addLast(new ServerHandler());
                        }
                    });

            // 绑定端口并启动服务
            ChannelFuture future = bootstrap.bind(PORT).sync();
            logger.info("========Netty server started on port {}", PORT);

            // 等待服务端监听端口关闭
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            logger.error("Server interrupted", e);
        } finally {
            // 优雅关闭线程组
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            logger.info("====Netty server shutdown");
        }
    }

    /**
     * 服务端业务处理器
     */
    private static class ServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            logger.info("====Client connected: {}", ctx.channel().remoteAddress());
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            // 客户端断开连接时从map中移除
            String clientId = getClientId(ctx);
            if (clientId != null) {
                clientMap.remove(clientId);
                logger.info("====Client {} disconnected", clientId);
            }
            ctx.close();
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (!(msg instanceof ProtocolMsg)) {
                logger.warn("Received unknown message type: {}", msg.getClass().getName());
                return;
            }

            ProtocolMsg message = (ProtocolMsg) msg;
            byte command = message.getCommand();

            switch (command) {
                case Command.LOGIN:
                    handleLogin(ctx, message);
                    break;
                case Command.LOGOUT:
                    handleLogout(ctx, message);
                    break;
                case Command.MESSAGE:
                    handleMessage(ctx, message);
                    break;
                case Command.HEARTBEAT:
                    handleHeartbeat(ctx, message);
                    break;
                default:
                    logger.warn("Unknown command: {}", command);
                    break;
            }
        }

        private void handleLogin(ChannelHandlerContext ctx, ProtocolMsg protocol) {
            try {
                // 反序列化登录数据
                LoginRequest loginRequest = JsonUtil.fromJson(protocol.getData(), LoginRequest.class);

                // 模拟登录验证
                if (!"token188".equals(loginRequest.getToken())) {
                    sendResponse(ctx, Command.LOGIN_RESPONSE, new LoginResponse(false, "Invalid token"));
                    ctx.close();
                    return;
                }
                // 添加到客户端map
                clientMap.put(loginRequest.getUserId(), ctx);

                // 发送登录成功响应
                sendResponse(ctx, Command.LOGIN_RESPONSE, new LoginResponse(true, "Login success"));

                logger.info("====用户：{} 登录系统", loginRequest.getUserId());
            } catch (Exception e) {
                logger.error("Login error", e);
                sendResponse(ctx, Command.LOGIN_RESPONSE, new LoginResponse(false, "Login error"));
                ctx.close();
            }
        }

        private void handleLogout(ChannelHandlerContext ctx, ProtocolMsg protocol) {
            String clientId = getClientId(ctx);
            if (clientId != null) {
                clientMap.remove(clientId);
                logger.info("User {} logged out", clientId);
            }
            sendResponse(ctx, Command.LOGOUT_RESPONSE, new LogoutResponse(true));
            ctx.close();
        }

        private void handleMessage(ChannelHandlerContext ctx, ProtocolMsg protocol) {
            String clientId = getClientId(ctx);
            if (clientId == null) {
                sendResponse(ctx, Command.ERROR_RESPONSE, new ErrorResponse("您尚未登录"));
                return;
            }
            try {
                MessageRequest messageRequest = JsonUtil.fromJson(protocol.getData(), MessageRequest.class);

                // 检查目标用户是否存在
                ChannelHandlerContext targetCtx = clientMap.get(messageRequest.getToUserId());
                if (targetCtx == null) {
                    sendResponse(ctx, Command.MESSAGE_RESPONSE, new MessageResponse(false, "聊天对象不在线"));
                    return;
                }

                // 转发消息给目标用户
                sendResponse(targetCtx, Command.MESSAGE, new MessageForward(clientId, messageRequest.getContent()));

                // 发送成功响应给发送者
                sendResponse(ctx, Command.MESSAGE_RESPONSE, new MessageResponse(true, "Message send success."));

                logger.info("Message forwarded from {} to {}", clientId, messageRequest.getToUserId());
            } catch (Exception e) {
                logger.error("Message handling error", e);
                sendResponse(ctx, Command.ERROR_RESPONSE, new ErrorResponse("Message error"));
            }
        }

        private void handleHeartbeat(ChannelHandlerContext ctx, ProtocolMsg protocol) {
            logger.debug("Received heartbeat from {}", getClientId(ctx));
            sendResponse(ctx, Command.HEARTBEAT_RESPONSE, new HeartbeatResponse());
        }

        private void sendResponse(ChannelHandlerContext ctx, byte command, Object data) {
            try {
                byte[] bytes = JsonUtil.toJson(data).getBytes(StandardCharsets.UTF_8);
                ProtocolMsg response = new ProtocolMsg((byte) 1, (byte) 0, command, bytes.length, bytes);
                ctx.writeAndFlush(response);
            } catch (Exception e) {
                logger.error("Send response error", e);
            }
        }

        /**
         * 用于处理用户自定义事件（非普通I/O事件）。当以下情况发生时被触发：
         * 与 IdleStateHandler 配合实现心跳检测
         * 处理自定义用户事件（如业务超时事件）
         */
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state() == IdleState.READER_IDLE) {
                    // 读空闲处理（如关闭连接）
                    logger.warn("No heartbeat received from {}, closing connection", getClientId(ctx));
                    ctx.close();
                }
                /*else if (event.state() == IdleState.WRITER_IDLE) {
                    // 写空闲处理（如发送心跳包）
                    ctx.writeAndFlush(new HeartbeatRequest());
                }*/
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            logger.error("Channel error: {}", getClientId(ctx), cause);
            ctx.close();
        }

        private String getClientId(ChannelHandlerContext ctx) {
            return clientMap.entrySet().stream().filter(entry -> entry.getValue().equals(ctx))
                    .map(Map.Entry::getKey)
                    .findFirst()
                    .orElse(null);
        }
    }
}
